package com.qushiyun.devops.admin.sync.schedule;

import com.qushiyun.devops.admin.sync.data.mapper.TriggerJobMapper;
import com.qushiyun.devops.admin.sync.data.po.TriggerJob;
import com.qushiyun.devops.admin.sync.job.SyncTemplate;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * 定时任务处理
 *
 * @author zhouzx
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class JobScheduleService {

    private final SyncTemplate syncTask;
    private final TriggerJobMapper triggerJobMapper;

    protected boolean started = false;
    protected volatile boolean offering = false;

    Worker worker = new Worker();

    BlockingQueue<TriggerJob> blockingQueue;

    public static final int DEFAULT_QUEUE_SIZE = 256;
    int queueSize = DEFAULT_QUEUE_SIZE;

    public static final int DEFAULT_MAX_FLUSH_TIME = 1000;
    int maxFlushTime = DEFAULT_MAX_FLUSH_TIME;

    @PostConstruct
    public void init() {
        start();
        List<TriggerJob> jobs = triggerJobMapper.getScheduleJob();
        jobs.forEach(blockingQueue::offer);
    }


    public void start() {
        if (isStarted()) {
            return;
        }
        blockingQueue = new ArrayBlockingQueue<>(queueSize);
        worker.setDaemon(true);
        worker.setName("job-schedule");
        worker.start();
        started = true;
    }

    @PreDestroy
    public void stop() {
        if (!isStarted()) {
            return;
        }
        worker.interrupt();
        try {
            worker.join(maxFlushTime);
            if (worker.isAlive()) {
                log.warn("Max queue flush timeout (" + maxFlushTime + " ms) exceeded. Approximately " + blockingQueue.size()
                    + " queued events were possibly discarded.");
            } else {
                log.info("Queue flush finished successfully within timeout.");
            }

        } catch (InterruptedException e) {
            log.error("", e);
        }
    }

    public boolean isStarted() {
        return started;
    }

    public void offer(TriggerJob schedule) {
        offering = true;
        boolean removed = blockingQueue.remove(schedule);
        boolean offered = false;
        if (schedule.getTriggerType() == 2) {
            offered = blockingQueue.offer(schedule);
        }
        log.info("add success {}", removed && offered);
        offering = false;
    }

    class Worker extends Thread {

        public void run() {
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                //ignore
            }
            while (isStarted()) {
                try {
                    if (offering) {
                        continue;
                    }
                    TriggerJob job = blockingQueue.take();
                    if (System.currentTimeMillis() - 5 >= job.getTriggerTimestamp()) {
                        try {
                            syncTask.execute(job.getId());
                        } catch (Exception e) {
                            log.warn("execute job error", e);
                        }
                    } else {
                        blockingQueue.offer(job);
                    }
                } catch (InterruptedException ie) {
                    break;
                }
            }

            for (TriggerJob job : blockingQueue) {
                try {
                    syncTask.execute(job.getId());
                } catch (Exception e) {
                    log.warn("execute job error", e);
                }
            }
        }
    }

}
